package trident.aggregation;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.trident.TridentTopology;
import org.apache.storm.trident.testing.FixedBatchSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;
import trident.partition.PrintFunction;

public class AggregationDemo {
    public static void main(String[] args) throws Exception {
        FixedBatchSpout spout = new FixedBatchSpout(new Fields("name", "hive", "storm"), 4,
                new Values("A", 80, 60),
                new Values("V", 80, 40),
                new Values("C", 50, 80),
                new Values("B", 70, 60),
                new Values("E", 80, 90),
                new Values("W", 30, 50),
                new Values("M", 90, 60),
                new Values("K", 60, 90),
                new Values("S", 70, 50)
        );

        TridentTopology tridentTopology = new TridentTopology();
//        tridentTopology.newStream("spout", spout)
//                .partitionAggregate(new Fields("hive"), new HiveCombinerAggregator(), new Fields("hiveSum"))
//                .each(new Fields("hiveSum"), new PrintFunction(), new Fields());
//        tridentTopology.newStream("spout", spout)
//                .partitionAggregate(new Fields("storm"), new StormReducerAggregator(), new Fields("stormSum"))
//                .each(new Fields("stormSum"), new PrintFunction(), new Fields());
//        tridentTopology.newStream("spout", spout)
//                .partitionAggregate(new Fields("hive", "storm"), new SumAggregation(), new Fields("sumState"))
//                .each(new Fields("sumState"), new PrintFunction(), new Fields());
        tridentTopology.newStream("spout", spout)
                .chainedAgg()
                .partitionAggregate(new Fields("hive"), new HiveCombinerAggregator(), new Fields("hiveSum"))
                .partitionAggregate(new Fields("storm"), new StormReducerAggregator(), new Fields("stormSum"))
                .chainEnd()
                .each(new Fields("hiveSum", "stormSum"), new PrintFunction(), new Fields())
        ;

        Config config = new Config();
        LocalCluster localCluster = new LocalCluster();
        localCluster.submitTopology("aggregation-demo", config, tridentTopology.build());
    }
}
